/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2016-2026 hmc,Inc.All Rights Reserved.
 */
package com.sys.midware.MQ.rocketMQ.logic.impl;

import java.util.concurrent.atomic.AtomicLong;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import com.sys.midware.MQ.rocketMQ.logic.IProducer;
import com.sys.midware.MQ.rocketMQ.model.MsgSendResult;
import com.sys.midware.assist.utils.string.StringUtil;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;


/**
 * 发送scd消息的实现
 * <b>ClassName：</b> ProducerImpl <br/>
 * <b>Description：</b> 发送scd消息的实现 <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2015年10月19日 下午2:04:20 <br/>
 * <b>@version: </b>  <br/>
 */
public class ProducerImpl implements IProducer {

	private static final  Logger logger = LoggerFactory.getLogger(ProducerImpl.class);
	public AtomicLong count = new AtomicLong();
	private final DefaultMQProducer producer;

	public ProducerImpl(String producerGroup) {
		this(null, producerGroup);
	}

	/**
	 * 构造函数
	 * @param producerGroup
	 *            实例组名字
	 * @param nameServer
	 *            nameServer 服务器地址
	 */
	public ProducerImpl( String nameServer,String producerGroup) {
	    this(nameServer,null,producerGroup);
	}
	   /**
     * 构造函数
     * @param producerGroup
     *            实例组名字
     * @param nameServer
     *            nameServer 服务器地址
     */
    public ProducerImpl( String nameServer,String instance,String producerGroup) {
        
        if(instance ==null){
            instance = producerGroup;
        }
        producer = new DefaultMQProducer(producerGroup);
        producer.setInstanceName(instance);
        producer.setNamesrvAddr(nameServer);
        
    }
    @Override
    public MsgSendResult sendMsg(String topic, String scdMsg) {
       return sendMsg(topic,null,scdMsg);
    }
	
	@Override
	public MsgSendResult sendMsg(String topic, String tags, String scdMsg) {
	    return sendMsg(topic,tags,null,scdMsg);
	}
	
	   public MsgSendResult sendMsg(String topic, String tags,String keys, String msgBody) {
	       MsgSendResult msgSendResult = new MsgSendResult();
	        try {
	            Message msg = new Message(topic, msgBody.getBytes());
	            msg.setWaitStoreMsgOK(false);
	            if(StringUtil.isNotEmpty(tags)){
	                msg.setTags(tags);
	            }
	            if(StringUtil.isNotEmpty(keys)){
	                msg.setKeys(keys);
	            }
	            SendResult sendResult = producer.send(msg);
	            msgSendResult.setSendResult(sendResult);
	            if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
	                msgSendResult.setSuccess(false);
	                return msgSendResult;
	            }
	            if (logger.isDebugEnabled()) {
	                logger.debug("send===========================>"
	                        + count.incrementAndGet());
	            }
	        } catch (Exception e) {
	            logger.error("send  msg on error",e);
	            msgSendResult.setSuccess(false);
	            return msgSendResult;
	        }
	        msgSendResult.setSuccess(true);
	        return msgSendResult;
	    }

	@Override
	public void shutdown() {
		producer.shutdown();
	}

    @Override
    public void start()  {
        try {
            logger.info(" ============= producer 准备启动 ==========");
            producer.start();
            logger.info(" nameServer "+producer.getNamesrvAddr() +",instanceName" + producer.getInstanceName() +" , group "+producer.getProducerGroup());
            logger.info(" ============= producer 启动成功 ==========");
        } catch (MQClientException e) {
            logger.error("============= Producer 启动失败 ==========", e);
        }
    }

    /* (non-Javadoc)
     * @see com.xuanyan.hmc.midware.MQ.rocketMQ.logic.IProducer#sendMsgWithKeys(java.lang.String, java.lang.String, java.lang.String)
     */
    @Override
    public MsgSendResult sendMsgWithKeys(String topic, String keys, String msgContent) {
        return sendMsg(topic,null,keys,msgContent);
    }
}
